package yarnrpc

import java.util.UUID

import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory


class MyNodeManager(val resourcemanager_hostname: String,
                    val resourcemanager_port: Int,
                    val nodemanager_cpu: Int,
                    val nodemanager_mem: Int) extends Actor {

  var nodemanagerid: String = _
  var rmRef :ActorSelection= _

  override def preStart(): Unit = {
    rmRef = context.actorSelection(s"akka.tcp://${Constant.RESOURCE_MANAGER_ACTOR_SYSTEM}@${resourcemanager_hostname}:${resourcemanager_port}/user/${Constant.RESOURCE_MANAGER_ACTOR}")
    nodemanagerid = UUID.randomUUID().toString
    rmRef ! RegisterNodeManager(nodemanagerid, nodemanager_cpu, nodemanager_mem)
  }

  def handleRegisteredNodeManager(rmhostname: String): Unit = {
    println(s"${rmhostname}:发送注册成功消息")
    //sender() ! HeartBeat(nodemanagerid)
    import scala.concurrent.duration._
    import context.dispatcher
    context.system.scheduler.schedule(0 millis, Constant.NODE_MANAGER_SEND_HEARTBEAT_TIME millis, context.self, SendHeartBeatToSelf())
  }

  override def receive: Receive = {
    case RegisteredNodeManager(rmhostname) => {
      handleRegisteredNodeManager(rmhostname)
    }
    case SendHeartBeatToSelf =>{
      rmRef ! HeartBeat(nodemanagerid)
      println(s"发送心跳 id:${nodemanagerid}")
    }
    case SendHeartBeatToSelf() =>{
      rmRef ! HeartBeat(nodemanagerid)
      println(s"发送心跳 id:${nodemanagerid}")
    }
    case _ => {
        println("不应该进入到这里")
      }
  }
}

object MyNodeManager {
  def apply(resourcemanager_hostname: String, resourcemanager_port: Int, nodemanager_cpu: Int, nodemanager_mem: Int):
  MyNodeManager =
    new MyNodeManager(resourcemanager_hostname, resourcemanager_port, nodemanager_cpu, nodemanager_mem)

  def main(args: Array[String]): Unit = {
    val RESOURCEMANAGER_HOSTNAME = args(0) // resource manager  主机名
    val RESOURCEMANAGER_PORT = args(1).toInt // resource manager 端口号
    val NODEMANAGER_HOSTNAME = args(2) // node manager 主机名
    val NODEMANAGER_PORT = args(3).toInt // node manager 指定的端口号 随意指定
    val NODEMANAGER_CPU = args(4).toInt //  node manager cpu 核数
    val NODEMANAGER_MEM = args(5).toInt // node manager mem 内存总量
    println(s"ResourceManager 主机名:${RESOURCEMANAGER_HOSTNAME} 端口: ${RESOURCEMANAGER_PORT}")
    println(s"本机 主机名:${NODEMANAGER_HOSTNAME} 端口:${NODEMANAGER_PORT} CPU:${NODEMANAGER_CPU} MEM:${NODEMANAGER_MEM}")
    val str =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname =${NODEMANAGER_HOSTNAME}
         |akka.remote.netty.tcp.port=${NODEMANAGER_PORT}
      """.stripMargin
    val config = ConfigFactory.parseString(str)
    val nodemanageractorsystem = ActorSystem(Constant.NODE_MANAGER_ACTOR_SYSTEM, config)
    nodemanageractorsystem.actorOf(Props(MyNodeManager(RESOURCEMANAGER_HOSTNAME, RESOURCEMANAGER_PORT, NODEMANAGER_CPU, NODEMANAGER_MEM)), Constant.NODE_MANAGER_ACTOR)
  }
}
